package com.liuqi.knet.remoting.transport.netty;

import com.liuqi.knet.Constants;
import com.liuqi.knet.remoting.Channel;
import com.liuqi.knet.remoting.ChannelHandler;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.ServerConfig;
import com.liuqi.knet.remoting.transport.AbstractServer;
import com.liuqi.knet.remoting.transport.dispatcher.DispatcherHandlers;
import com.liuqi.knet.remoting.transport.netty.logging.NettyHelper;
import com.liuqi.knet.util.NetUtil;
import com.liuqi.loggers.api.Logger;
import com.liuqi.loggers.facade.LoggersFacade;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;

/**
 * @author liuqi
 * @date 2025/2/8 16:59
 **/
public class NettyServer extends AbstractServer {

    static {
        NettyHelper.setNettyLoggerFactory();
    }

    private static final Logger logger = LoggersFacade.getLogger(NettyServer.class);

    private ServerBootstrap bootstrap;
    private io.netty.channel.Channel serverChannel;

    private Map<String, Channel> channels; // <ip:port, channel>

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    private NettyCodecAdapter nettyCodecAdapter;

    public NettyServer(ChannelHandler handler, ServerConfig config) throws RemotingException {
        super(new DispatcherHandlers(handler, config).warpServer(), config);
    }

    @Override
    protected void doBind() throws Throwable {
        final NettyServerHandler nettyServerHandler = new NettyServerHandler(getChannelHandler(), getNetConfig());
        channels = nettyServerHandler.getChannels();

        bootstrap = new ServerBootstrap();

        nettyCodecAdapter = new NettyCodecAdapter(getCodec());
        bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(Constants.serverBossGroupName, true));
        workerGroup = new NioEventLoopGroup(Constants.workerThreads,
                new DefaultThreadFactory(Constants.serverWorkerGroupName, true));

        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast("decoder", nettyCodecAdapter.getDecoder())
                                .addLast("encoder", nettyCodecAdapter.getEncoder())
                                .addLast("handler", nettyServerHandler);
                    }
                });

        logger.info("开始启动服务端: " + getLocalAddress());

        // bind
        ChannelFuture channelFuture = bootstrap.bind(getLocalAddress());
        channelFuture.syncUninterruptibly();
        serverChannel = channelFuture.channel();
    }

    @Override
    protected void doClose() throws Throwable {
        try {
            // close all client connect
            Collection<Channel> channels = getChannels();
            if (channels != null && channels.size() > 0) {
                for (Channel channel : channels) {
                    try {
                        channel.close();
                    } catch (Throwable e) {
                        logger.warn(e.getMessage(), e);
                    }
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

        // 关闭本服务端
        try {
            if (serverChannel != null) {
                serverChannel.close();
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

        // group.shutdown
        try {
            if (bootstrap != null) {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public boolean isBound() {
        return serverChannel.isActive();
    }

    @Override
    public Collection<Channel> getChannels() {
        Collection<Channel> chs = new HashSet<Channel>();
        for (Channel channel : this.channels.values()) {
            if (channel.isActive()) {
                chs.add(channel);
            } else {
                channels.remove(NetUtil.toAddressString(channel.getRemoteAddress()));
            }
        }
        return chs;
    }

    @Override
    public Channel getChannel(InetSocketAddress remoteAddress) {
        return channels.get(NetUtil.toAddressString(remoteAddress));
    }
}
